package repair

import (
	"center/db"
	"center/db/models"
	"common/node_pieces"
	"common/utils"
	"common/uuid"
	"context"
	"fmt"
	"github.com/wuzfei/go-helper/slices"
	"github.com/zeebo/errs"
	"go.uber.org/zap"
	"golang.org/x/sync/errgroup"
	"os"
	"path/filepath"
	"strconv"
	"time"
)

var Error = errs.Class("Repair")
var ErrorScanOver = Error.New("scan over")

type Config struct {
	Internal     time.Duration `help:"扫描表间隔时间" default:"10m"`
	ScanLimit    int           `help:"每次扫表行数" default:"1000"`
	MaxWaitTime  time.Duration `help:"等待修复最长时间" default:"30m"`
	AfterOffTime time.Duration `help:"获取在线节点时，在这个掉线时间内的也获取" default:"30m"`
	DebugLastId  int64         `help:"从某个id开始，调使用，上线置为0" default:"0"`
}

type Service struct {
	config Config
	log    *zap.Logger
	db     *db.CenterDB
	lastId int64
}

func NewService(log *zap.Logger, db *db.CenterDB, conf Config) *Service {
	return &Service{
		config: conf,
		log:    log,
		db:     db,
	}
}

func (s *Service) Run(ctx context.Context) error {
	if s.config.DebugLastId > 0 {
		s.lastId = s.config.DebugLastId
	} else {
		s.lastId = getLastId()
	}

	eg := errgroup.Group{}
	eg.Go(func() error {
		// processWaitRepair 处理修复等待时间过长的记录，有可能节点异常，未将节点报上来
		return utils.RunTickerAfter(ctx, s.config.Internal, func() error {
			err := s.db.ResetRepairWaitTooLong(ctx, s.config.MaxWaitTime)
			if err != nil {
				s.log.Error("processWaitRepair error ", zap.Error(err))
			}
			return nil
		})
	})
	eg.Go(func() error {
		return utils.RunTickerAfter(ctx, s.config.Internal, func() error {
			err := s.worker(ctx)
			//扫表结束
			//if errors.Is(err, ErrorScanOver) {
			//	time.Sleep(s.config.Internal)
			//	continue
			//}
			s.log.Error("修复扫表完成结果：", zap.Error(err))
			return nil
		})
	})
	return eg.Wait()
}

func (s *Service) worker(ctx context.Context) (err error) {
	onlineNodes, err := s.getOnlineNodeIdxs(ctx)
	if err != nil {
		return err
	}
	for {
		select {
		case <-ctx.Done():
			return nil
		default:
			err = s.process(ctx, onlineNodes)
			if errs.Is(err, ErrorScanOver) {
				s.lastId = s.config.DebugLastId
			}
			setLastId(s.lastId)
			if err != nil {
				return err
			}
		}
	}
}

func (s *Service) process(ctx context.Context, onlineNodes []uint32) (err error) {
	s.log.Info("repair process", zap.Time("t", time.Now()), zap.Int64("id", s.lastId))
	segs, err := s.db.GetSegmentsForRepairCheck(ctx, s.lastId, s.config.ScanLimit)
	if err != nil {
		return err
	}

	if len(segs) == 0 {
		return ErrorScanOver
	}

	streamIds := make([]uuid.UUID, 0, len(segs))
	for _, seg := range segs {
		streamIds = append(streamIds, seg.StreamId)
		if s.lastId < seg.ID {
			s.lastId = seg.ID
		}
	}

	//获取未过期的文件，过期的就不用修复了
	objects, err := s.db.GetObjectsByStreamIds(ctx, streamIds)
	if err != nil {
		return err
	}
	objectMap := make(map[string]*models.Object)
	for _, o := range objects {
		//过期就删除文件
		if o.Expired() {
			_err := s.db.DeleteObjectByStreamId(ctx, o.StreamId)
			if _err != nil {
				s.log.Error("删除文件失败（DeleteObjectByStreamId）", zap.ByteString("streamId", o.StreamId.Bytes()))
			}
			continue
		}
		objectMap[o.StreamId.String()] = o
	}
	if len(objectMap) == 0 {
		s.log.Info("objectMap empty ")
		return nil
	}

	repairQueue := make([]*models.Repair, 0)
	for _, seg := range segs {
		s.lastId = seg.ID
		if seg.Status != models.StatusNormal {
			continue
		}
		l := uint32(len(compareOnlineNodeId(seg.NodePieces, onlineNodes)))
		if o, ok := objectMap[seg.StreamId.String()]; ok {
			if l >= o.Redundancy.Required && l <= o.Redundancy.Repair {
				s.log.Debug("添加修复segment",
					zap.Int64("segmentId", seg.ID),
					zap.String("l", fmt.Sprintf("%d <= %d <= %d", o.Redundancy.Required, l, o.Redundancy.Repair)))
				repairQueue = append(repairQueue, &models.Repair{
					SegmentID:  seg.ID,
					StreamId:   seg.StreamId,
					Position:   seg.Position,
					Status:     models.RepairStatusWaiting,
					Redundancy: o.Redundancy,
				})
			} else if l < o.Redundancy.Required {
				s.log.Warn("piece数量不够，无法修复", zap.String("", seg.StreamId.String()),
					zap.Uint32("pieces count", l), zap.String("nodePieces", seg.NodePieces.String()))
			}
		} else {
			s.log.Error("StreamId not int objectMap")
		}
	}
	return s.db.CreateRepairs(ctx, repairQueue)
}

func compareOnlineNodeId(node node_pieces.NodePieces, online []uint32) []uint32 {
	res := make([]uint32, 0, len(node))
	if len(node) == 0 || len(online) == 0 {
		return res
	}
	for _, n := range node {
		for _, o := range online {
			if n == o {
				res = append(res, n)
				continue
			}
		}
	}
	return res
}

func (s *Service) getOnlineNodeIds(ctx context.Context) ([]uuid.UUID, error) {
	nodes, err := s.db.GetAllOnlineNodeAfterTime(ctx, s.config.AfterOffTime)
	if err != nil {
		return nil, err
	}
	return slices.Map(nodes, func(v *models.Node, k int) uuid.UUID {
		return v.ID
	}), nil
}

func (s *Service) getOnlineNodeIdxs(ctx context.Context) ([]uint32, error) {
	nodes, err := s.db.GetAllOnlineNodeAfterTime(ctx, s.config.AfterOffTime)
	if err != nil {
		return nil, err
	}
	return slices.Map(nodes, func(v *models.Node, k int) uint32 {
		return v.IDx
	}), nil
}

var lastIdFile = filepath.Join(os.TempDir(), "estore_last_id")

func setLastId(id int64) {
	_ = os.WriteFile(lastIdFile, []byte(strconv.Itoa(int(id))), 0766)
}

func getLastId() int64 {
	r, err := os.ReadFile(lastIdFile)
	if err != nil {
		return 0
	}
	i, err := strconv.ParseInt(string(r), 10, 64)
	if err != nil {
		return 0
	}
	return i
}
